package com.study.flink.demo;

import java.util.PriorityQueue;
import java.util.UUID;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.StringUtils;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestDemo {
    OutputTag<String> some = new OutputTag<String>("sdf");
    public void run() {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> control = env.fromElements("DROP", "IGNORE");
        try {
            control.print();
            env.execute();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class ConnectedCarEvent {
        String carId;
        Long timestamp;
        
        public ConnectedCarEvent() {
        }

        public ConnectedCarEvent(String carId, Long timestamp) {
            super();
            this.carId = carId;
            this.timestamp = timestamp;
        }

        public static ConnectedCarEvent random() {
            return new ConnectedCarEvent(UUID.randomUUID().toString(), System.currentTimeMillis());
        }

        public static ConnectedCarEvent fromString(String line) {
            ConnectedCarEvent event = null;
            // .....
            return event;
        }
    }
    
    public static class SortFunction extends KeyedProcessFunction<String, ConnectedCarEvent, ConnectedCarEvent> {
        private static final long serialVersionUID = 1L;

        /* we'll use a PriorityQueue to buffer not-yet-fully-sorted events */
        private ValueState<PriorityQueue<ConnectedCarEvent>> queueState = null;

        @Override
        public void open(Configuration config) {
            /* set up the state we want to use */
            ValueStateDescriptor<PriorityQueue<ConnectedCarEvent>> descriptor = new ValueStateDescriptor<>(
                "sorted-events", TypeInformation.of(new TypeHint<PriorityQueue<ConnectedCarEvent>>() {})
            );
            queueState = getRuntimeContext().getState(descriptor);
        }

        @Override
        public void processElement(ConnectedCarEvent event, Context context, Collector<ConnectedCarEvent> out) throws Exception {
            /* add/sort this event into the queue */ 
            TimerService timerService = context.timerService();

            if (context.timestamp() > timerService.currentWatermark()) {
                PriorityQueue<ConnectedCarEvent> queue = queueState.value();
                if (queue == null) {
                    queue = new PriorityQueue<>(10);
                }
                queue.add(event);
                queueState.update(queue);
                /* set an event-time timer for when the stream is complete up to the event-time of this event */
                timerService.registerEventTimeTimer(event.timestamp);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext context, Collector<ConnectedCarEvent> out) throws Exception {
            /* release the items at the head of the queue that are now ready, based on the CurrentWatermark */
            PriorityQueue<ConnectedCarEvent> queue = queueState.value();
            Long watermark = context.timerService().currentWatermark();
            ConnectedCarEvent head = queue.peek();
            while (head != null && head.timestamp <= watermark) {
                out.collect(head);
                queue.remove(head);
                head = queue.peek();
            }
        }
    }

    private void some() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(4);

        DataStreamSource<TaskDTO> dtoStream = env.addSource(new TaskSource());
        KeyedStream<TaskDTO, Tuple> keyedStream = dtoStream.keyBy("tasksEnums");

        DataStream<TaskDTO> filtedStream = keyedStream.filter(new FilterFunction<TaskDTO>() {
            @Override
            public boolean filter(TaskDTO task) throws Exception {
                return null != task && StringUtils.isNullOrWhitespaceOnly(task.getUuid());
            }
        });
        
        filtedStream.map(line -> line + "1");

        DataStream<Tuple2<String, Integer>> mapedStream = filtedStream
            .map(new RichMapFunction<TaskDTO, Tuple2<String, Integer>>() {
                private ValueState<Integer> counts;

                @Override
                public void open(Configuration parameters) throws Exception {
                    counts = getRuntimeContext()
                        .getState(new ValueStateDescriptor<Integer>("typeCounts", Integer.class));
                }

                @Override
                public Tuple2<String, Integer> map(TaskDTO task) throws Exception {
                    Integer thisCount = counts.value();

                    if(null == thisCount) {
                        thisCount = 0;
                    }

                    counts.update(++thisCount);
                    return new Tuple2<String, Integer>(task.getTasksEnums().toString(), thisCount);
                }
            });

        mapedStream.writeAsCsv("/data/output/tasks.csv");
        env.execute();
    }
    
    public void broadcast() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.enableCheckpointing(1000);
        env.setParallelism(4);

        // 任务流
        DataStreamSource<TaskDTO> dtoStream = env.addSource(new TaskSource());
        // 任务推送的规则流，该流目前还不是广播流。
        DataStreamSource<PushMsgConf> pushMsgStream = env.addSource(new PushMsgSource());

        KeyedStream<TaskDTO, String> keyedStream = dtoStream
            .keyBy(new KeySelector<TaskDTO, String>() {
                @Override
                public String getKey(TaskDTO value) throws Exception {
                    return value.getTasksEnums();
                }
            });

        // 创建一个Map描述，存放任务类型和推送配置。
        MapStateDescriptor<String, PushMsgConf> ruleStateDescriptor = new MapStateDescriptor<>(
            "RulesBroadcastState", BasicTypeInfo.STRING_TYPE_INFO,
            TypeInformation.of(new TypeHint<PushMsgConf>() {
            }));

        // 将“任务推送的规则流”转换为广播流。此时会创建 Broadcast State。
        BroadcastStream<PushMsgConf> ruleBroadcastStream = pushMsgStream
            .broadcast(ruleStateDescriptor);

        SingleOutputStreamOperator<TaskDTO> stream = keyedStream
            // 链接任务流和广播流，返回值为 BroadcastConnectedStream 。
            .connect(ruleBroadcastStream)
            // 如果是 KeyedStream 需使用接口 KeyedBroadcastProcessFunction ；
            // 如果是 non-keyed stream 需要使用接口 BroadcastProcessFunction 。
            .process(new KeyedBroadcastProcessFunction<String, TaskDTO, PushMsgConf, TaskDTO>() {
                private final MapStateDescriptor<String, PushMsgConf> ruleStateDescriptor = new MapStateDescriptor<>(
                    "RulesBroadcastState", BasicTypeInfo.STRING_TYPE_INFO,
                    TypeInformation.of(new TypeHint<PushMsgConf>() {
                    }));

                // 这里的逻辑必须是“确定性”的，不要加入“随机因子”。因为Flink需要保证所有平行实例中“广播结果”的一致性。
                @Override
                public void processBroadcastElement(PushMsgConf value,
                    KeyedBroadcastProcessFunction<String, TaskDTO, PushMsgConf, TaskDTO>.Context ctx,
                    Collector<TaskDTO> out) throws Exception {
                    // 存储广播数据到MapState。
                    ctx.getBroadcastState(ruleStateDescriptor).put(value.getTasksEnums(), value);
                }

                @Override
                public void processElement(TaskDTO value,
                    KeyedBroadcastProcessFunction<String, TaskDTO, PushMsgConf, TaskDTO>.ReadOnlyContext ctx,
                    Collector<TaskDTO> out) throws Exception {
                    // 获取广播以来的配置
                    ReadOnlyBroadcastState<String, PushMsgConf> configs = ctx
                        .getBroadcastState(ruleStateDescriptor);
                    // 根据配置处理任务流的任务。
                    // ....
                }
            });

        stream.writeAsCsv("/data/output/tasks.csv");
        env.execute();
    }
    
    class TaskDTO {
        public String getUuid() {
            return "";
        }
        public String getTasksEnums() {
            return "";
        }
    }
    
    class TaskDO {
        public String uuid;
        public String tasksEnums;
        
        public String getUuid() {
            return "";
        }
        public String getTasksEnums() {
            return "";
        }
    }
    
    class PushMsgConf {
        public String tasksEnums;
        public String channel;

        public String getTasksEnums() {
            return tasksEnums;
        }
        public String getChannel() {
            return channel;
        }
    }
    
    class TaskSource implements SourceFunction<TaskDTO> {

        @Override
        public void run(SourceContext<TaskDTO> ctx) throws Exception {
        }

        @Override
        public void cancel() {
        }
        
    }
    
    class PushMsgSource implements SourceFunction<PushMsgConf> {
        @Override
        public void run(SourceContext<PushMsgConf> ctx) throws Exception {
        }

        @Override
        public void cancel() {
        }
    }
}
